package com.cet.billingjob.config;

import com.cet.billingjob.domain.BillingData;
import com.cet.billingjob.domain.ReportData;
import com.cet.billingjob.exception.PricingException;
import com.cet.billingjob.jobs.BillingDataProcessor;
import com.cet.billingjob.jobs.FilePreparationTasklet;
import com.cet.billingjob.jobs.FileParameterValidator;
import com.cet.billingjob.listener.BillingDataSkipListener;
import com.cet.billingjob.listener.ReportDataRetryListener;
import com.cet.billingjob.service.PricingService;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.FlatFileItemWriter;
import org.springframework.batch.item.file.FlatFileParseException;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.builder.FlatFileItemWriterBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;
import org.springframework.jdbc.core.DataClassRowMapper;
import org.springframework.jdbc.support.JdbcTransactionManager;

import javax.sql.DataSource;

/**
 * @author 30952
 */

@Configuration
public class BillingJobConfig {

    @Bean
    public Job job(JobRepository jobRepository, Step step1, Step step2, Step step3) {
        return new JobBuilder("billingJob", jobRepository)
            .validator(new FileParameterValidator())
            .start(step1)
            .next(step2)
            .next(step3)
            .build();
    }

    @Bean
    public Step step1(JobRepository jobRepository, @Qualifier("batchTransactionManager") JdbcTransactionManager jdbcTransactionManager) {
        return new StepBuilder("filePreparation", jobRepository)
            .tasklet(new FilePreparationTasklet(), jdbcTransactionManager)
            .build();
    }

    @Bean
    public Step step2(JobRepository jobRepository,
                      @Qualifier("batchTransactionManager") JdbcTransactionManager jdbcTransactionManager,
                      ItemReader<BillingData> billingFlatFileItemReader,
                      ItemWriter<BillingData> billingJdbcBatchItemWriter,
                      BillingDataSkipListener skipListener) {
        return new StepBuilder("fileIngesting", jobRepository)
            .<BillingData, BillingData> chunk(100, jdbcTransactionManager)
            .reader(billingFlatFileItemReader)
            .writer(billingJdbcBatchItemWriter)
            .faultTolerant()
            .skip(FlatFileParseException.class)
            .skipLimit(10)
            .listener(skipListener)
            .build();
    }

    @Bean
    public Step step3(JobRepository jobRepository,
                      @Qualifier("batchTransactionManager") JdbcTransactionManager jdbcTransactionManager,
                      ItemReader<BillingData> billingDataJdbcCursorItemReader,
                      ItemProcessor<BillingData, ReportData> billingDataProcessor,
                      ItemWriter<ReportData> billingDataFileWriter,
                      ReportDataRetryListener retryListener) {
        return new StepBuilder("reportGeneration", jobRepository)
            .<BillingData, ReportData> chunk(100, jdbcTransactionManager)
            .reader(billingDataJdbcCursorItemReader)
            .processor(billingDataProcessor)
            .writer(billingDataFileWriter)
            .faultTolerant()
            .retry(PricingException.class)
            .retryLimit(100)
            .listener(retryListener)
            .build();
    }

    @Bean
    @StepScope
    public FlatFileItemReader<BillingData> billingFlatFileItemReader(@Value("#{jobParameters['input.file']}") String inputFile) {
        return new FlatFileItemReaderBuilder<BillingData>()
            .name("billingDataFileReader")
//            .resource(new FileSystemResource("staging/billing-2023-01.csv"))
            .resource(new FileSystemResource(inputFile))
            // 默认使用','分割
            .delimited()
            .names("dataYear", "dataMonth", "accountId", "phoneNumber", "dataUsage", "callDuration", "smsCount")
            .targetType(BillingData.class)
            .build();
    }

    @Bean
    public JdbcBatchItemWriter<BillingData> billingJdbcBatchItemWriter(@Qualifier("batchDataSource") DataSource dataSource) {

        String sql = "insert into billing_data values(:dataYear, :dataMonth, :accountId, :phoneNumber, :dataUsage, :callDuration, :smsCount)";
        return new JdbcBatchItemWriterBuilder<BillingData>()
            .dataSource(dataSource)
            .sql(sql)
            // 将javaBean中的属性绑定到相同的sql占位符中， :value
            .beanMapped()
            .build();
    }

    @Bean
    @StepScope
    public JdbcCursorItemReader<BillingData> billingDataJdbcCursorItemReader(@Qualifier("batchDataSource") DataSource dataSource,
                                                                             @Value("#{jobParameters['data.year']}") Integer year,
                                                                             @Value("#{jobParameters['data.month']}") Integer month) {
        String sql = String.format("select * from billing_data where data_year = %d and data_month = %d", year, month);
        return new JdbcCursorItemReaderBuilder<BillingData>()
            .name("billingDataTableReader")
            .dataSource(dataSource)
            .sql(sql)
            .rowMapper(new DataClassRowMapper<>(BillingData.class))
            .build();
    }

    @Bean
    public BillingDataProcessor billingDataProcessor(PricingService pricingService) {
        return new BillingDataProcessor(pricingService);
    }

    @Bean
    @StepScope
    public FlatFileItemWriter<ReportData> billingDataFileWriter(@Value("#{jobParameters['output.file']}") String outputFile) {
        return  new FlatFileItemWriterBuilder<ReportData>()
            .name("billingDataFileWriter")
//            .resource(new FileSystemResource("staging/billing-report-2023-01.csv"))
            .resource(new FileSystemResource(outputFile))
            .delimited()
            .delimiter("|")
            .names("billingData.dataYear", "billingData.dataMonth", "billingData.accountId", "billingData.phoneNumber", "billingData.dataUsage", "billingData.callDuration",
                "billingData.smsCount", "billingTotal")
            .build();
    }

    @Bean
    @StepScope
    public BillingDataSkipListener skipListener(@Value("#{jobParameters['skip.file']}") String skippedFile) {
        return new BillingDataSkipListener(skippedFile);
    }

    @Bean
    public ReportDataRetryListener retryListener() {
        return new ReportDataRetryListener();
    }

    @Bean
    public PricingService pricingService() {
        return new PricingService();
    }
}
